package consumer

import (
	"code.bydev.io/cht/customer/kyc_admin_service/internal/model"
	"code.bydev.io/frameworks/byone/core/jsonx"
	"context"
	"emperror.dev/errors"
	"encoding/json"
	"time"

	"code.bydev.io/cht/customer/kyc-stub.git/pkg/bybit/kyc/common/enums"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/constant"
	kycdm "code.bydev.io/cht/customer/kyc_admin_service/internal/domain/kyc"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/awss3"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/config"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/helper"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycadminapi"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycadminevent"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycadminevent/dto"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycadminrepo"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycprovider"
	kpdto "code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/kycprovider/dto"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/monitor"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/pkg/utils/timeutil"
	"code.bydev.io/cht/customer/kyc_admin_service/internal/utils/ukafka"
	"code.bydev.io/cht/customer/kyc_dbstructure.git/pkg/o_kyc_action_image"
	"code.bydev.io/frameworks/byone/core/logc"
	"code.bydev.io/frameworks/byone/kafka"
	"code.bydev.io/frameworks/sechub-go/api"
	"git.bybit.com/svc/go/pkg/bdal"
	"git.bybit.com/svc/mod/pkg/bsecret"
	pbkyc "git.bybit.com/svc/stub/pkg/pb/api/kyc"
	"git.bybit.com/svc/stub/pkg/pb/enums/ekycadmin"
	"git.bybit.com/svc/stub/pkg/pb/enums/kyc"
	jsoniter "github.com/json-iterator/go"
	"go.uber.org/zap"
)

type Domain struct {
	logger         *zap.Logger
	tell           kycadminapi.Teller
	repo           kycadminevent.KycNotifyRepo
	kycSecHub      *api.Sechub
	chWorker       chan bool
	kycProducer    *ukafka.ProducerClient
	kycActionState *model.KycActionState
}

func NewDomain(
	secHub *api.Sechub,
	chtKycDB *bdal.Environment,
	userDB *bdal.Environment,
	kfk kafka.Producer,
	logger *zap.Logger,
	teller kycadminapi.Teller,
	kycSecHub *api.Sechub,
	kycSvc kycdm.KycSvc,
	internalKey bsecret.DataKey,
	kycProducer *ukafka.ProducerClient,
) *Domain {
	repo := kycadminrepo.New(logger, chtKycDB, userDB, secHub, teller, kycSvc, internalKey, kfk)

	workerNum := config.GetInt(config.KeyWorkerNum)
	if workerNum <= 0 || workerNum > 10 {
		workerNum = 10
	}

	ch := make(chan bool, workerNum)
	return &Domain{
		repo:           repo,
		logger:         logger,
		tell:           teller,
		kycSecHub:      kycSecHub,
		chWorker:       ch,
		kycProducer:    kycProducer,
		kycActionState: new(model.KycActionState),
	}
}

func (kac *Domain) ByoneConsumer(ctx context.Context, msg []byte) {
	logc.Infow(ctx, "ByoneConsumer begin", logc.Field("msg", string(msg)))
	kac.chWorker <- true
	go kac.ConsumeWorker(ctx, msg)
	logc.Infow(ctx, "ByoneConsumer end")
}

func (kac *Domain) ConsumeWorker(ctx context.Context, msg []byte) {
	logc.Infow(ctx, "ConsumeWorker begin", logc.Field("msg", string(msg)))
	defer func(ch chan bool) {
		<-ch
	}(kac.chWorker)

	// 处理逻辑
	ctxTimeOut, cancel := context.WithTimeout(ctx, 2*time.Minute)
	defer cancel()
	_, err := kac.notifyHandler(ctxTimeOut, msg)
	if err != nil {
		logc.Errorw(ctx, "kac.NotifyHandler error", logc.Field("err", err))
		// 增加告警
		_ = kac.tell.KycAlarm(ctxTimeOut, "kac.NotifyHandler error: "+string(msg), err)
		return
	}
	logc.Infow(ctx, "ConsumeWorker end")
}

func (kac *Domain) notifyHandler(ctx context.Context, notify []byte) (bool, error) {
	logc.Infow(ctx, "notifyHandler begin", logc.Field("notify", string(notify)))
	metrics := monitor.NewMetrics("SumSubCompensate", "")
	kycNotify, err := kac.parseKYCEventNotify(ctx, notify)
	if err != nil {
		metrics.AddMetrics(err, "parseKYCEventNotify")
		return false, err
	}
	logc.Infow(ctx, "kac.parseKYCEventNotify info", logc.Field("kycNotify", kycNotify))

	if kycNotify.EventType == ekycadmin.KYCAdminAction_ACTION_Level2STATECOMPLETE {
		logc.Infow(ctx, "KYC event KYCAdminAction_ACTION_Level2STATECOMPLETE")
		return true, nil
	}

	// 过滤kyc信息转移消息
	if kycNotify.EventType == ekycadmin.KYCAdminAction_ACTION_TRANSFER {
		logc.Infow(ctx, "KYC event is identify transfer")

		// 上报目标用户的信息到filecore kafka
		if kycNotify.Level != 0 { // 当前消息的用户是转移的目标用户
			memberID := kycNotify.MemberID

			// 生成基础证件的上传filecore任务
			onboardingLv1Record, err := kac.kycActionState.GetLatestApprovedOnboardPOIRecord(ctx, memberID)
			if err != nil {
				logc.Errorw(ctx, "GetLatestApprovedOnboardPOIRecord Err", logc.Field("err", err))
				return false, err
			}
			if onboardingLv1Record == nil {
				logc.Errorw(ctx, "approvedOnboardPOIRecord should not be nil",
					logc.Field("kycNotify", kycNotify))
				return false, errors.New("approvedOnboardPOIRecord shoud not be nil")
			}

			if err = kac.filecoreUploadNotify(ctx, memberID, onboardingLv1Record.IdempotentID, kpdto.ImgTypeDocFront); err != nil {
				logc.Errorw(ctx, "filecoreUploadNotify failed",
					logc.Field("kycNotify", kycNotify),
					logc.Field("onboardingLv1Record", onboardingLv1Record),
				)
				return false, errors.New("filecoreUploadNotify failed")
			}
			if err = kac.filecoreUploadNotify(ctx, memberID, onboardingLv1Record.IdempotentID, kpdto.ImgTypeDocBack); err != nil {
				logc.Errorw(ctx, "filecoreUploadNotify failed",
					logc.Field("kycNotify", kycNotify),
					logc.Field("onboardingLv1Record", onboardingLv1Record),
				)
				return false, errors.New("filecoreUploadNotify failed")
			}
			if err = kac.filecoreUploadNotify(ctx, memberID, onboardingLv1Record.IdempotentID, kpdto.ImgTypeFace); err != nil {
				logc.Errorw(ctx, "filecoreUploadNotify failed",
					logc.Field("kycNotify", kycNotify),
					logc.Field("onboardingLv1Record", onboardingLv1Record),
				)
				return false, errors.New("filecoreUploadNotify failed")
			}

			// 生成补充证件的上传filecore任务
			supplementLv1Records, err := kac.kycActionState.GetAllApprovedSupplementPOIRecord(ctx, memberID)
			if err != nil {
				logc.Errorw(ctx, "GetAllApprovedSupplementPOIRecord Err", logc.Field("err", err))
				return false, err
			}
			for _, v := range supplementLv1Records {
				if err = kac.filecoreUploadNotify(ctx, memberID, v.IdempotentID, kpdto.ImgTypeDocFront); err != nil {
					logc.Errorw(ctx, "filecoreUploadNotify failed",
						logc.Field("kycNotify", kycNotify),
						logc.Field("SupplementLv1Record", v),
					)
					return false, errors.New("filecoreUploadNotify failed")
				}
				if err = kac.filecoreUploadNotify(ctx, memberID, v.IdempotentID, kpdto.ImgTypeDocBack); err != nil {
					logc.Errorw(ctx, "filecoreUploadNotify failed",
						logc.Field("kycNotify", kycNotify),
						logc.Field("SupplementLv1Record", v),
					)
					return false, errors.New("filecoreUploadNotify failed")
				}
				if err = kac.filecoreUploadNotify(ctx, memberID, v.IdempotentID, kpdto.ImgTypeFace); err != nil {
					logc.Errorw(ctx, "filecoreUploadNotify failed",
						logc.Field("kycNotify", kycNotify),
						logc.Field("SupplementLv1Record", v),
					)
					return false, errors.New("filecoreUploadNotify failed")
				}
			}
		}

		return true, nil
	}
	logc.Infow(ctx, "ActionStateHandler begin")
	tag, err := kac.repo.ActionStateHandler(ctx, kycNotify)
	if err != nil {
		metrics.AddMetrics(err, "ActionStateHandler")
		logc.Errorw(ctx, "ActionStateHandler error", logc.Field("err", err))
		return false, err
	}

	// Lv2 终审通过流水处理成功时补发等级变更通知
	if tag && kycNotify.EventType == ekycadmin.KYCAdminAction_ACTION_COMPLETE && kycNotify.Level == 2 && kycNotify.ResultDetail.FinalDecision == 1 {
		errLv2Resend := kac.resendLevel2Notify(ctx, kycNotify.MemberID, 2, kpdto.ReviewResult(kycNotify.ResultDetail.FinalDecision))
		if errLv2Resend != nil {
			logc.Errorw(ctx, "r.resendLevel2Notify error", logc.Field("err", errLv2Resend))
		}
		logc.Infow(ctx, "resendLevel2Notify success", logc.Field("ntf", kycNotify))
	}

	// 补偿证件优先下载
	_ = kac.SelfieUpload4Supplement(ctx, kycNotify)

	metrics.AddMetrics(nil, monitor.END)
	logc.Infow(ctx, "notifyHandler end")
	return tag, nil
}

func (kac *Domain) parseKYCEventNotify(ctx context.Context, notify []byte) (*dto.KYCEventNotify, error) {
	rsp := new(dto.KYCEventNotify)
	err := json.Unmarshal(notify, rsp)
	if err != nil {
		logc.Errorw(ctx, "json.Unmarshal error", logc.Field("err", err))
		return nil, err
	}
	logc.Infow(ctx, "get KYCEventNotify info", logc.Field("rsp", rsp))

	et, _ := timeutil.FloatMsStringToUnix(rsp.EventTime)
	rt, _ := timeutil.FloatMsStringToUnix(rsp.ResultDetail.ReviewTime)

	// 由于 KYC 事件的 provider 定义是：ProviderSumSub=0 ProviderJumio=1
	// 为了确保数据库、接口、pb 文件一致，这里做一下转换
	rsp.Provider = kyc.KYCProvider(int(rsp.Provider) + 1)
	// 由于 KYC 事件的 event_type 定义不一样，需要做一下转换
	kac.convEventType(rsp)

	rsp.EventTimeMs = et
	rsp.ResultDetail.ReviewTimeMs = rt

	// Jumio 的时候需要查询补齐 recordId
	if rsp.Provider == kyc.KYCProvider_PROVIDER_JUMIO && rsp.RecordID == "" {
		userAccountID, userDataID, err := kac.repo.GetJumioWrkFlwID(ctx, rsp.MemberID)
		if err != nil {
			logc.Infow(ctx, "kac.repo.GetJumioWrkFlwID error", logc.Field("err", err))
		} else {
			rsp.AccountID = userAccountID
			rsp.RecordID = userDataID
		}
	} else if rsp.Provider == kyc.KYCProvider_PROVIDER_ONFIDO && rsp.RecordID == "" {
		comm, err := kac.repo.GetKYCRecordCommon(ctx, rsp.MemberID)
		if err != nil {
			logc.Infow(ctx, "kac.repo.GetKYCRecordCommon error", logc.Field("err", err))
		} else {
			rsp.AccountID = comm.ApplicantId
			rsp.RecordID = comm.DataId
		}
	} else if rsp.Provider == kyc.KYCProvider_PROVIDER_SUMSUB && rsp.RecordID == "" {
		rsp.RecordID = rsp.AccountID
	}

	// Token 事件处理下 Business ID
	if rsp.EventType == ekycadmin.KYCAdminAction_ACTION_INIT && rsp.BusinessID == "" {
		rsp.BusinessID = constant.BusinessTypeGlobal.String()
	}

	logc.Infow(ctx, "get new KYCEventNotify info", logc.Field("rsp", rsp))
	// 处理 extra
	kac.extraDataHandler(ctx, rsp)

	return rsp, nil
}

func (kac *Domain) SelfieUpload4Supplement(ctx context.Context, notify *dto.KYCEventNotify) error {
	if notify == nil {
		return nil
	}
	if notify.LevelName == kpdto.POISupplementPreDoc.String() || notify.LevelName == kpdto.POISupplement.String() {
		record, errSlf := kac.repo.GetSelfieRecord(ctx, notify.MemberID)
		if errSlf != nil {
			logc.Errorw(ctx, "GetSelfieRecord error", logc.Field("err", errSlf))
			return errSlf
		}
		if record == nil {
			logc.Infow(ctx, "GetSelfieRecord is nil", logc.Field("notify", notify))
			return nil
		}
		err := kac.handleImage(ctx, record)
		if err != nil {
			logc.Errorw(ctx, "handleImage error", logc.Field("record", record), logc.Field("err", err))
			return err
		}
	}
	return nil
}

func (kac *Domain) extraDataHandler(ctx context.Context, rsp *dto.KYCEventNotify) {
	logc.Infow(ctx, "extraDataHandler begin", logc.Field("rsp", rsp))
	if rsp.ExtraData.RawResponse != "" {
		rrspStr, err := helper.DeSecHubCodeString(kac.kycSecHub, rsp.ExtraData.RawResponse)
		if err != nil {
			logc.Errorw(ctx, "helper.DeSecHubCodeString error", logc.Field("err", err))
			return
		}
		rrsp := &dto.RawResponse{}
		err = jsoniter.UnmarshalFromString(rrspStr, &rrsp)
		if err != nil {
			logc.Errorw(ctx, "jsoniter.UnmarshalFromString error", logc.Field("err", err))
			return
		}
		logc.Infow(ctx, "get rrsp info", logc.Field("rrsp", rrsp))
		rsp.RawResponse.One = rrsp.One
		rsp.RawResponse.Status = rrsp.Status
	}
	logc.Infow(ctx, "extraDataHandler end", logc.Field("rsp", rsp))
}

func (kac *Domain) convEventType(rsp *dto.KYCEventNotify) {
	ket := int32(rsp.EventType)
	if (ket >= int32(dto.Init) && ket <= int32(dto.Reset)) ||
		(ket == int32(dto.IdentifyTransfer) || ket == int32(dto.Level2StateComplete)) {
		ket--
		rsp.EventType = ekycadmin.KYCAdminAction(ket)
	} else {
		// 对应 KYC 的 Amend 事件
		rsp.IsAmend = true
		rsp.EventType = ekycadmin.KYCAdminAction_ACTION_RESET
	}
}

func (kac *Domain) handleImage(ctx context.Context, record *o_kyc_action_image.Record) error {
	logc.Infow(ctx, "handleImage begin")
	// 如果不存在，取得圖片並上傳
	actionImg := &kpdto.KYCActionImage{
		IdempotentID: record.IdempotentId,
		MemberID:     record.MemberId,
		Provider:     enums.KYCProvider(record.Provider),
		AccountID:    record.AccountId,
		RecordID:     record.RecordId,
		ImgID:        record.ImgId,
		ImgType:      kpdto.ImgType(record.ImgType),
	}
	// 取得圖片並上傳
	doc, err := kac.getImageRawAndUpload(ctx, actionImg)
	if err != nil {
		logc.Errorw(ctx, "s.GetImageRawAndUpload error", logc.Field("err", err))
		return err
	}
	logc.Infow(ctx, "s.GetImageRawAndUpload", logc.Field("doc", doc))
	// 更新 DB
	err = kac.repo.ActionImageComplete(ctx, record, doc)
	if err != nil {
		logc.Errorw(ctx, "ActionImageComplete error", logc.Field("err", err))
		return err
	}
	logc.Infow(ctx, "handleImage end")

	// 挂载上传至filecore的生产者
	err = kac.filecoreUploadNotify(ctx, actionImg.MemberID, actionImg.IdempotentID, actionImg.ImgType)
	if err != nil {
		logc.Errorw(ctx, "filecoreUploadNotify error", logc.Field("err", err))
		return err
	}

	return nil
}

func (kac *Domain) getImageRawAndUpload(ctx context.Context, actionImg *kpdto.KYCActionImage) (*kpdto.Document, error) {
	logc.Infow(ctx, "getImageRawAndUpload begin")
	kycProvider := kycprovider.NewProvider(kac.logger, kpdto.KYCProvider(actionImg.Provider), kac.tell)
	s3 := awss3.New(kac.logger)
	// 取得圖片
	doc, err := kycProvider.GetImageRaw(ctx, actionImg)
	if err != nil {
		logc.Errorw(ctx, "GetImageRaw error", logc.Field("err", err))
		return nil, err
	}
	if doc == nil {
		logc.Errorw(ctx, "GetImageRaw failed", logc.Field("doc", doc))
		return nil, nil
	}
	// 上傳圖片
	err = s3.Upload(ctx, actionImg.IdempotentID, doc)
	if err != nil {
		logc.Errorw(ctx, "Upload error", logc.Field("err", err))
		return nil, err
	}
	logc.Infow(ctx, "getImageRawAndUpload end")
	return doc, nil
}

// resendLevel2Notify 补发等级变更通知
func (kac *Domain) resendLevel2Notify(ctx context.Context, memberID, level int64, result kpdto.ReviewResult) error {
	bytes, _ := jsonx.Marshal(&dto.KYCResultNotify{
		MemberID: memberID,
		LeveL:    level,
		Result:   result,
	})
	err := kac.kycProducer.KYCResult.Send(ctx, &kafka.Message{Value: bytes})
	if err != nil {
		logc.Errorw(ctx, "sendLevel2Notify error", logc.Field("err", err))
	}

	return err
}

// filecoreUploadNotify 图片文件上传filecore任务
func (kac *Domain) filecoreUploadNotify(ctx context.Context, memberID int64, idempotentID string, imgType kpdto.ImgType) error {
	logc.Infow(ctx, "filecoreUploadNotify",
		logc.Field("memberID", memberID),
		logc.Field("idempotentID", idempotentID),
		logc.Field("imgType", imgType),
	)

	if imgType < kpdto.ImgTypeDocFront || imgType > kpdto.ImgTypeFace {
		logc.Errorw(ctx, "invalid imgType", logc.Field("idempotentID", idempotentID), logc.Field("imgType", imgType))
		return nil
	}

	// 当前只允许issue country == TUR 的身份证, 可以上传到s3 (满足Narkasa photoSharing诉求)
	// 当前只允许证件正反面、自拍上传
	record, err := kac.repo.GetActionByIdempotentID(ctx, idempotentID)
	if err != nil {
		logc.Errorw(ctx, "GetActionByIdempotentID error", logc.Field("err", err))
		return err
	}

	if !(record.Decision == 1 && record.Country == "TUR" && record.DocType == 1) {
		logc.Infow(ctx, "not allowed to upload to filecore",
			logc.Field("idempotentID", idempotentID),
			logc.Field("actionRecord", record),
		)
		return nil
	}

	fileBizType := pbkyc.FileBizType(imgType + 1)
	bytes, _ := json.Marshal(&dto.FilecoreUploadNotify{
		MemberID:     memberID,
		IdempotentID: idempotentID,
		FileBizType:  fileBizType,
	})
	err = kac.kycProducer.FilecoreUpload.Send(ctx, &kafka.Message{Value: bytes})
	if err != nil {
		logc.Errorw(ctx, "send FilecoreUploadNotify error", logc.Field("err", err))
		return err
	}

	logc.Infow(ctx, "filecoreUploadNotify Success",
		logc.Field("memberID", memberID),
		logc.Field("idempotentID", idempotentID),
		logc.Field("imgType", imgType),
	)
	return nil
}
